package com.li.Listener;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @program: srpingboot-rocketmq
 * @description: 生产者 消息事务监听
 * @author: li
 * @create: 2022-09-26 11:25
 **/
@RocketMQTransactionListener(corePoolSize = 10,
        maximumPoolSize = 10,
        keepAliveTime = 10L,
        rocketMQTemplateBeanName = "rocketMQTemplate")
public class MqTransactionListener implements RocketMQLocalTransactionListener {

    private AtomicInteger transactionIndex = new AtomicInteger(0);
    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
    // git 1

    /**
     * 执行本地事务逻辑
     *
     * @param msg
     * @param arg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        System.out.println("MqTransactionListener executeLocalTransaction msg :" + msg.toString());
        System.out.println("MqTransactionListener executeLocalTransaction arg :" + arg.toString());
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put(arg.toString(), status);
        RocketMQLocalTransactionState state = getRocketMQLocalTransactionState(status);
        System.out.println("MqTransactionListener executeLocalTransaction status :" + state.name());
        return state;
    }

    /**
     * 回查本地二次确认结果
     *
     * @param msg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        System.out.println("MqTransactionListener checkLocalTransaction msg :" + msg.toString());
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put("MqTransactionListener checkLocalTransaction_" + System.currentTimeMillis(), status);
        RocketMQLocalTransactionState transactionState = getRocketMQLocalTransactionState(status);
        System.out.println("MqTransactionListener checkLocalTransaction status :" + transactionState.name());
        return transactionState;
    }


    private RocketMQLocalTransactionState getRocketMQLocalTransactionState(Integer status) {
        switch (status) {
            case 0:
                // 返回中间状态， broker会重新检查 消息发送着的二次确认状态(commit或rollback)
                return RocketMQLocalTransactionState.UNKNOWN;
            case 1:
                // 提交事务，broker 将半事务消息 改为可投掷
                return RocketMQLocalTransactionState.COMMIT;
            case 2:
                // 回滚事务， broker 将该半事务删除,不允许消费
                return RocketMQLocalTransactionState.ROLLBACK;

        }
        return RocketMQLocalTransactionState.UNKNOWN;
    }
}
